package com.juma.cartridge.retryque;

import java.util.concurrent.DelayQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import lombok.extern.slf4j.Slf4j;

import com.juma.cartridge.retryque.dao.RetryQueueDao;

/**
 * 	任务续租线程
 * @ClassName TaskLeaseRrolongKeepper
 * @Description
 * @Author zhuangguangyou
 * @Date 2019年6月18日 下午5:43:54
 * @Version 1.0
 *
 */

@Slf4j
public class LeaseProlongKeepper implements Runnable {

	private RetryQueueDao taskQueueDao;
	private AtomicBoolean running;
	private DelayQueue<TaskLeaseProlong> leaseQueue;
	
	public LeaseProlongKeepper(DelayQueue<TaskLeaseProlong> leaseQueue,RetryQueueDao dao) {
		this.leaseQueue = leaseQueue;
		this.taskQueueDao = dao;
		this.running = new AtomicBoolean(true);
	}
	
	public void close() {
		running.set(false);
	}
	
	
	@Override
	public void run() {
		log.debug("任务续租线程启动...");
		while(running.get() && !Thread.currentThread().isInterrupted()) {
			try {
			    TaskLeaseProlong prolong = leaseQueue.take();
			    if(prolong.getIds().isEmpty()) {
			        continue;
			    }
			    log.debug("取到续租任务[leaseVal={}]",prolong.getProlongVal());
                int n = taskQueueDao.prolongLock(prolong);
                if(n>0) {
                    prolong.incrLeaseVal();
                    leaseQueue.add(prolong); //重新放入续租队列
                    log.debug("本次成功续租{}条任务",n);    
                }
			}catch(InterruptedException ie) {    
				break;
			}
		}
		log.debug("任务续租线程结束!");
	}
	
}
